package com.suxin.mqtt.util;

import com.suxin.mqtt.domain.MqAggregation;
import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
import org.eclipse.paho.client.mqttv3.MqttCallback;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.springframework.scheduling.concurrent.CustomizableThreadFactory;

import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 * @Classname Consumer
 * @Description [ TODO ]
 * @Author Tang
 * @Date 2020/12/1 14:58
 * @Created by ASUS
 */
public class AbstractConsumer implements MqttCallback {

    private IHandler iHandler;

    private ThreadPoolExecutor executor;

    public AbstractConsumer(IHandler iHandler) {
        this.iHandler = iHandler;
        executor = new ThreadPoolExecutor(Runtime.getRuntime().availableProcessors(),
                Runtime.getRuntime().availableProcessors() + 1,50, TimeUnit.MILLISECONDS
                ,new LinkedBlockingQueue<Runnable>(16),new CustomizableThreadFactory("mq-thread-"));
    }

    @Override
    public void connectionLost(Throwable throwable) {
        // 连接丢失后，一般在这里面进行重连
        System.out.println("连接断开，可以做重连...");
    }

    @Override
    public void messageArrived(String topic, MqttMessage mqttMessage) throws Exception {
        // subscribe后得到的消息会执行到这里面
        String payload = new String(mqttMessage.getPayload());
        MqAggregation mqAggregation = new MqAggregation(topic,payload,mqttMessage);
        executor.execute(() -> {
            iHandler.handler(mqAggregation);
        });
    }

    @Override
    public void deliveryComplete(IMqttDeliveryToken iMqttDeliveryToken) {
        System.out.println("deliveryComplete---------" + iMqttDeliveryToken.isComplete());
    }
}